/*
 * @(#)CanalServer.java   1.0  2021年11月8日
 * 
 * Copyright (c)	2014-2020. All Rights Reserved.	GuangZhou hhmk Technology Company LTD.
 */
package com.swift.dts.source.canal;

import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.instance.manager.ManagerCanalInstanceGenerator;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.swift.core.env.ConfigInterface.ChangeConfigKV;
import com.swift.core.spring.SpringAutowireBeanFactory;
import com.swift.dts.source.BinlogEventSource;
import com.swift.exception.extend.SystemException;
import com.swift.util.type.TypeUtil;

/**
 * 添加说明 
 * @author zhengjiajin
 * @version 1.0 2021年11月8日
 */
@Component
public class CanalDtsServer extends CanalServerWithEmbedded {

    @Autowired
    private SpringAutowireBeanFactory springAutowireBeanFactory;
    
    @Autowired
    private LocalFileCanalConfig canalConfigClient;
    
    private ManagerCanalInstanceGenerator canalInstanceGenerator;
    

    private AtomicInteger clientId = new AtomicInteger(0);
    

    private Map<String, BinlogEventSource> clientThread = new java.util.concurrent.ConcurrentHashMap<>();
    
    @PostConstruct
    public void start() {
        canalInstanceGenerator = new ManagerCanalInstanceGenerator();
        canalInstanceGenerator.setCanalConfigClient(canalConfigClient);
        super.setCanalInstanceGenerator(canalInstanceGenerator);
        super.start();
    }
    
    @PreDestroy
    public void stop() {
        stopAllClient();
        super.stop();
    }
    
    public void stopAllClient() {
        for(String destination:clientThread.keySet()) {
            stopClient(destination);
        } 
    }
    
    public void restart(List<ChangeConfigKV<String>> changeList) {
        if (TypeUtil.isNull(changeList)) return;
        changeList.forEach(change -> {
            if (TypeUtil.isNull(change.getNewConfig())) {
                stopClient(change.getOldConfig());
            } else if (TypeUtil.isNull(change.getOldConfig())) {
                startClient(change.getNewConfig());
            } else {
                ClientIdentity client = stopClient(change.getOldConfig());
                startClient(client);
            }
        });

    }
    
    /*
     * 监听所有
     */
    /*
     * @PostConstruct private void init() { if (TypeUtil.isNull(canalConfigClient.getAllDestination())) return; for
     * (String destination : canalConfigClient.getAllDestination()) { startNew(destination); } }
     */

    public void startClient(String destination) {
        ClientIdentity clientIdentity = new ClientIdentity();
        clientIdentity.setClientId((short) clientId.incrementAndGet());
        clientIdentity.setDestination(destination);
        startClient(clientIdentity);
    }

    public void startClient(ClientIdentity clientIdentity) {
        synchronized (this) {
            super.start(clientIdentity.getDestination());
            super.subscribe(clientIdentity);
            BinlogEventSource thread = new BinlogEventSource(clientIdentity);
            springAutowireBeanFactory.autowire(thread);
            clientThread.put(clientIdentity.getDestination(), thread);
            thread.start();
        }
    }
    

    private ClientIdentity stopClient(String destination) {
        synchronized (this) {
            BinlogEventSource thread = clientThread.remove(destination);
            if (thread == null) return null;
            super.unsubscribe(thread.getClientIdentity());
            super.stop(destination);
            thread.setStop(true);
            return thread.getClientIdentity();
        }
    }
    
    @SuppressWarnings("rawtypes")
    private CanalEventParser getEventParser(String destination) {
        CanalInstance canalInstance = this.getCanalInstances().get(destination);
        if (canalInstance == null) throw new SystemException("canalInstance未注册");
        CanalEventParser canalEventParser = canalInstance.getEventParser();
        return canalEventParser;
    }

    @SuppressWarnings("rawtypes")
    public void resetBlackFilter(String blackFilter) {
        Map<String, CanalInstance> map = this.getCanalInstances();
        if (TypeUtil.isNull(map)) return;
        map.keySet().forEach(destination -> {
            AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(blackFilter);
            CanalEventParser canalEventParser = getEventParser(destination);
            if (canalEventParser instanceof AbstractEventParser) {
                ((AbstractEventParser) canalEventParser).setEventBlackFilter(aviaterFilter);
            }
            if (canalEventParser instanceof GroupEventParser) {
                GroupEventParser groupEventParser = (GroupEventParser) canalEventParser;
                for (CanalEventParser parser : groupEventParser.getEventParsers()) {
                    if (parser instanceof AbstractEventParser) {
                        ((AbstractEventParser) parser).setEventBlackFilter(aviaterFilter);
                    }
                }
            }

        });
    }

    @SuppressWarnings("rawtypes")
    public void resetFilter(String filter) {
        Map<String, CanalInstance> map = this.getCanalInstances();
        if (TypeUtil.isNull(map)) return;
        map.keySet().forEach(destination -> {
            AviaterRegexFilter aviaterFilter = new AviaterRegexFilter(filter);
            CanalEventParser canalEventParser = getEventParser(destination);
            if (canalEventParser instanceof AbstractEventParser) {
                ((AbstractEventParser) canalEventParser).setEventFilter(aviaterFilter);
            }
            if (canalEventParser instanceof GroupEventParser) {
                GroupEventParser groupEventParser = (GroupEventParser) canalEventParser;
                for (CanalEventParser parser : groupEventParser.getEventParsers()) {
                    if (parser instanceof AbstractEventParser) {
                        ((AbstractEventParser) parser).setEventFilter(aviaterFilter);
                    }
                }
            }
        });
    }
    
}
